package simple;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.HashSet;
import java.util.Set;

/**
 * 拉模式消费者
 */
public class PullConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer();
        consumer.setNamesrvAddr("");
        Set<String> topics = new HashSet<>();
        topics.add("Simple");
        topics.add("TopicTest");
        consumer.setRegisterTopics(topics);
        consumer.start();
        while (true){

            consumer.getRegisterTopics().forEach(n->{
                try{
                    Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues(n);
                    messageQueues.forEach(l->{
//                      获取偏移量 offset
                        try {
                            //1先从内存中读
                            long offset = consumer.getOffsetStore().readOffset(l, ReadOffsetType.READ_FROM_MEMORY);
                            //2再从‘l’队列中读
                            if (offset<0){
                                offset = consumer.getOffsetStore().readOffset(l,ReadOffsetType.READ_FROM_STORE);
                            }
                            //3读取‘l’队列中最大偏移量
                            if (offset<0){
                                offset = consumer.maxOffset(l);
                            }
                            if (offset<0){
                                offset = 0;
                            }
                            // 拉取消息pullResult，‘l’即每个消息队列，‘*’即接收所有，offset偏移量(消息是根据偏移量找到新消息的)，maxNums每次从’l‘消息队列中获取多少条消息
                            PullResult pullResult = consumer.pull(l, "*", offset, 32);
                            System.out.println("消息拉取成功_"+pullResult);

                            switch (pullResult.getPullStatus()){

//                              FOUND,找到了，NO_NEW_MSG,没找到，NO_MATCHED_MSG没有匹配的消息，OFFSET_ILLEGAL非法偏移量（if (offset<0){
//                                offset = 0;
//                            }）已经处理了，赋值为‘0’默认值
                                case FOUND:
                                    pullResult.getMsgFoundList().forEach(k->{
                                        System.out.println("消息消费成功"+k);
                                    });
                                    //更新偏移量(拿到下一个偏移量)
                                    consumer.updateConsumeOffset(l,pullResult.getNextBeginOffset());
                            }
                            } catch (MQClientException e) {
                                e.printStackTrace();
                            } catch (RemotingException e) {
                                e.printStackTrace();
                            } catch (MQBrokerException e) {
                                e.printStackTrace();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        });
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
            });
        }
    }
}
